package org.apache.spark

import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef, RpcEnv}
import org.apache.spark.sql.SparkSession

import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}

object RpcClientMain {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    val sparkSession = SparkSession.builder().config(conf).master("local[*]").appName("test rpc").getOrCreate()
    val sparkContext: SparkContext = sparkSession.sparkContext
    val sparkEnv: SparkEnv = sparkContext.env

    val rpcEnv: RpcEnv = RpcEnv
      .create(HelloRpcSettings.getName(),
        HelloRpcSettings.getHostname(),
        HelloRpcSettings.getPort(),
        conf,
        sparkEnv.securityManager, false)

    //RpcEndpointRef 相当于akka系统中的一个ActorRef，通过这个组件和其他组件通信
    val endPointRef: RpcEndpointRef = rpcEnv
      .setupEndpointRef(RpcAddress(HelloRpcSettings.getHostname(), HelloRpcSettings.getPort()), HelloRpcSettings.getName())

    import scala.concurrent.ExecutionContext.Implicits.global

    //send 发送消息后就不管了
    endPointRef.send(SayHi("test send"))


    //ask 发送后异步等待对方返回结果
    val future: Future[String] = endPointRef.ask[String](SayHi("test ask"))

    //获取返回结果
    future.onComplete{
      case scala.util.Success(value)=>{
        println(s"Got the Ask result = $value")
      }
      case scala.util.Failure(e)=>{
        println(s"Got the Ask error: $e")
      }
    }

    Await.result(future, Duration.apply("30s"))


    //askSync 阻塞等待对方返回结果
    val res = endPointRef.askSync[String](SayBye("test askSync"))
    println(res)
    sparkSession.stop()


  }

}
